package com.atguigu.flink.wordcount;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2023/2/23
 *
 *   语法:  (参数列表) -> { 方法体 }
 *
 *      Collector<WordCount>中的WordCount称为泛型类型，缺失了。
 *              使用泛型的情况，lamda表达式不会提供足够的类型信息。
 *
 *             如何解决:  ①不用lamda，而用匿名内部类
 *                       ②坚持用lamda，需要明确地显式声明类型信息
 *                              在函数返回值的结尾部分，是哟个returns()给出缺失泛型的类型提示
 *
 *      Caused by: org.apache.flink.api.common.functions.InvalidTypesException:
 *              The generic type parameters of 'Collector' are missing.
 *              In many cases lambda methods don't provide enough information for automatic type extraction
 *              when Java generics are involved.
 *              An easy workaround is to use an (anonymous) class instead that implements the
 *              'org.apache.flink.api.common.functions.FlatMapFunction' interface.
 *              Otherwise the type has to be specified explicitly using type information.
 *
 *              Exception in thread "main" org.apache.flink.api.common.functions.InvalidTypesException:
 *                 The return type of function 'main(Demo7_LamdaPOJO.java:29)' could not be determined automatically,
 *                 due to type erasure. You can give type information hints by using the returns(...) method
 *                 on the result of the transformation call, or by letting your function implement the
 *                 'ResultTypeQueryable' interface.
 */
public class Demo7_LamdaPOJO
{
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);

        DataStreamSource<String> source = env.readTextFile("data/words.txt");


        SingleOutputStreamOperator<Demo5_POJO.WordCount> ds1 = source
            .flatMap((FlatMapFunction<String, Demo5_POJO.WordCount>) (line, out) -> {
                String[] words = line.split(" ");
                for (String word : words) {
                    out.collect(new Demo5_POJO.WordCount(word,1));
                }
            })
            //明确给出缺失的泛型的类型信息
            .returns(Demo5_POJO.WordCount.class);

        ds1
            // 方法引用。 lamda表达式的{}中的内容实现时一个已有的方法，可以使用 类名::方法名，直接引用
            // 使用POJO的某个字段作为key，都可以写 POJO::getXxx
            .keyBy(Demo5_POJO.WordCount::getWord)
            //写要sum的字段
            .sum("count")
            .print();

        env.execute();

    }


}
